package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.DorisUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2023/12/28
 * sku粒度下单聚合统计
 *      维度：sku
 *      度量：原始金额、优惠券减免金额、活动减免金额、实付金额
 * 需求启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、doris、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程
 *      环境准备
 *      从kafka的下单事实表主题中读取数据
 *      对读取的数据进行类型转换[并过滤空消息]  jsonStr->jsonObj
 *      去重
 *          为什么会产生重复数据？
 *              下单事实表是由订单明细、订单、订单明细活动以及订单明细优惠券4张表关联得到
 *              订单明细是主表
 *              和订单表关联使用的是内连接
 *              和订单明细活动以及订单明细优惠券表关联使用的是左外连接
 *              如果左外连接的时候，左表数据先到，右表数据后到，会产生3条数据
 *                  [左表     null]   +I
 *                  [左表     null]   -D
 *                  [左表     右表]   +I
 *              这样的数据发送到kafka主题中会接收到3条消息
 *                  左表  null
 *                  null
 *                  左表  右表
 *              当我们从kafka主题中读取数据的时候，如果使用的是FlinkSQL的方式，可以直接对空消息进行处理；
 *              如果使用的是FlinkAPI的方式，默认的SimpleStringSchema是处理不了空消息的，所以在MyKafkaUtil
 *              工具类的getKafkaSource方法中，需要自定义反序列化器的实现。
 *              空消息过滤掉后，剩下的2条数据属于重复数据，需要去重
 *          去重方式1：状态 + 定时器
 *              当数据到来的时候，先放到状态中，并注册5s后执行的定时器，
 *              如果有重复数据进来，会和状态中的数据的聚合时间进行比较，将聚合时间大的数据放到状态中
 *              当定时器被触发执行的时候，将状态的数据传递到下游
 *              不足：影响整个实时处理的时效性
 *          去重方式2：状态 + 抵消
 *              数据到来的时候，先放到状态中，并直接传递到下游
 *              如果有重复数据进来，会将状态中影响到度量值的属性进行取反，再传递到下游
 *              并且将后到来的时候，也传递到下游
 *              好处：时效性没有影响  不足：如果数据重复  会向下游发送3条数据
 *      指定Watermark以及提取事件时间字段
 *      再次对流中数据类型进行了转换  jsonObj->统计的实体类对象
 *      分组
 *      开窗
 *      聚合计算
 *      和维度进行关联
 *          基本的维度关联的实现
 *              HbaseUtil：  JSONObject getDimInfoFormHbase()
 *          优化1：旁路缓存
 *              思路：先从缓存中获取维度数据，如果从缓存中找到了要关联的维度，直接将其作为方法的返回值进行返回(缓存命中)
 *                   如果在缓存中没有找到要关联的维度，发送请求到Hbase找将维度数据查询出来，并将查询出的维度放到
 *                   缓存中缓存起来，方便下次查询使用
 *              缓存产品选型：
 *                  状态   :  性能很好    维护性差
 *                  redis ： 性能也不错   维护性好    √
 *              关于Redis的配置：
 *                  type:   String
 *                  key:    维度表名:主键值
 *                  TTL:    1day
 *                  注意：当维度数据发生了变化，需要将缓存的数据清除掉
 *              DimUtil:  JSONObject getDimInfo()
 *
 *              模板方法设计模式：
 *                  在父类中定义完成某一个功能的核心算法的骨架(步骤)，但是某些步骤在父类中实现不了，需要延迟到子类中
 *                  去实现。在不改变父类核心算法骨架的前提下，每一个子类都可以有自己不同的实现
 *                  DimMapFunction(模板)|DimJoinFunction(维度关联需要实现的方法)
 *
 *          优化2：异步IO
 *              为什么时候异步？
 *                  在对流中数据进行处理的时候，如果想提升处理能力，可以调大算子的并行度，但是更大的并行度意味着更多的
 *                  硬件资源，不可能无限制的调整；所以需要考虑，在硬件资源有限的情况下，去提升处理能力
 *                  在做维度关联的时候，我们使用的是map算子对流中数据进行处理，map算子在单个并行度上，使用的是同步的
 *                  方式处理流中数据，也就是说，处理完一个元素后才能处理下一个元素，等待的时间比较长，所以使用异步方式
 *                  进行优化。
 *              Flink提供的发送异步请求的API
 *                  AsyncDataStream.[un]orderedWait(
 *                      流,
 *                      如何发送异步请求(implements AsyncFunction),
 *                      超时时间,
 *                      时间单位
 *                  )
 *             我们将发送异步请求进行维度关联的操作抽取出了一个模板
 *             class DimAsyncFunction extends RichAsyncFunction implements DimJoinFunction{
 *                 open:初始化异步连接
 *                 close:关闭异步连接
 *                 asyncInvoke:发送异步请求
 *                      CompletableFuture
 *                          .supplyAsync        创建异步编排对象，执行异步任务--有返回值
 *                              发送请求到Redis获取维度数据
 *                          .thenApplyAsync     执行异步任务，有入参，有返回值
 *                              根据从Redis中查询到的维度情况，判断是否到Hbase中进行查询
 *                          .thenAcceptAsync    执行异步任务，有入参，无返回值
 *                              将维度属性补充到流中对象上并向下游传递
 *             }
 *             HbaseUtil添加了如下异步操作的方法:
 *                  获取异步连接
 *                  关闭异步连接
 *                  以异步的方式获取数据
 *             RedisUtil添加了如下异步操作的方法:
 *                  获取异步连接
 *                  关闭异步连接
 *                  以异步的方式获取数据
 *                  以异步的方式将数据放到Redis中
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关的设置
        env.enableCheckpointing(5000L);
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        //TODO 3.从kafka主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS
            = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //TODO 4.对流中的数据类型进行转换并过滤空消息  jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
            new ProcessFunction<String, JSONObject>() {
                @Override
                public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                    if (StringUtils.isNotEmpty(jsonStr)) {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        out.collect(jsonObj);
                    }
                }
            }
        );

        // jsonObjDS.print(">>>");
        //TODO 5.去重
        //5.1 按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //5.2 去重方案1 ：状态 + 定时器  不足：每来一条数据都要等5s后才能传递到下游，影响时效性
        /*SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastJsonObjState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastJsonObjState",JSONObject.class);
                    lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //从状态中获取数据
                    JSONObject lastJsonObj = lastJsonObjState.value();
                    if(lastJsonObj == null){
                        //说明现在没有出现重复
                        lastJsonObjState.update(jsonObj);
                        //注册一个5s钟之后执行的定时器
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                    }else{
                        //说明出现重复---伪代码
                        String ts1 = lastJsonObj.getString("聚合时间戳");
                        String ts2 = jsonObj.getString("聚合时间戳");
                        if(ts2.compareTo(ts1) >= 0){
                            lastJsonObjState.update(jsonObj);
                        }
                    }

                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject lastJsonObj = lastJsonObjState.value();
                    out.collect(lastJsonObj);
                    lastJsonObjState.clear();
                }
            }
        );*/
        //5.3 去重方案2 ：状态 + 抵消   优点：对失效性基本上没有影响   不足：如果出现重复了，要向下游传递3条数据，传递的数据变多
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastJsonObjState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastJsonObjState",JSONObject.class);
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                    lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //从状态中获取json
                    JSONObject lastJsonObj = lastJsonObjState.value();
                    if(lastJsonObj != null){
                        //说明重复了 将状态中影响到度量值的字段进行取反，再发送到下游，进行抵消
                        String splitOriginalAmount = lastJsonObj.getString("split_original_amount");
                        String splitCouponAmount = lastJsonObj.getString("split_coupon_amount");
                        String splitActivityAmount = lastJsonObj.getString("split_activity_amount");
                        String splitTotalAmount = lastJsonObj.getString("split_total_amount");
                        lastJsonObj.put("split_original_amount","-" + splitOriginalAmount);
                        lastJsonObj.put("split_coupon_amount","-" + splitCouponAmount);
                        lastJsonObj.put("split_activity_amount","-" + splitActivityAmount);
                        lastJsonObj.put("split_total_amount","-" + splitTotalAmount);
                        out.collect(lastJsonObj);
                    }
                    out.collect(jsonObj);
                    lastJsonObjState.update(jsonObj);
                }
            }
        );

        // distinctDS.print(">>>");

        //TODO 6.指定Watermark的生成策略以及提取事件时间字段
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = distinctDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<JSONObject>() {
                        @Override
                        public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                            return jsonObj.getLong("ts") * 1000;
                        }
                    }
                )
        );
        //TODO 7.再次对流中数据类型进行转换  jsonObj->统计的实体类对象
        //{"create_time":"2023-12-27 10:47:07","sku_num":"1","activity_rule_id":"1","split_original_amount":"6499.0000",
        // "split_coupon_amount":"0.0","sku_id":"3","date_id":"2023-12-27","user_id":"1710",
        // "province_id":"10","activity_id":"1","sku_name":"小米12S ","id":"14257139",
        // "order_id":"54586","split_activity_amount":"500.0","split_total_amount":"5999.0","ts":"1703731627"}
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanDS = withWatermarkDS.map(
            new MapFunction<JSONObject, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                    String skuId = jsonObj.getString("sku_id");
                    String splitOriginalAmount = jsonObj.getString("split_original_amount");
                    String splitCouponAmount = jsonObj.getString("split_coupon_amount");
                    String splitActivityAmount = jsonObj.getString("split_activity_amount");
                    String splitTotalAmount = jsonObj.getString("split_total_amount");
                    Long ts = jsonObj.getLong("ts") * 1000;
                    return TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .originalAmount(new BigDecimal(splitOriginalAmount))
                        .couponAmount(new BigDecimal(splitCouponAmount))
                        .activityAmount(new BigDecimal(splitActivityAmount))
                        .orderAmount(new BigDecimal(splitTotalAmount))
                        .ts(ts)
                        .build();
                }
            }
        );

        // orderBeanDS.print(">>>>");

        //TODO 8.按照统计的维度进行分组(sku_id)
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS = orderBeanDS.keyBy(TradeSkuOrderBean::getSkuId);

        //TODO 9.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
            = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 10.聚合计算
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                    value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                    value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                    value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                    value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                    return value1;
                }
            },
            new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                    String stt = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());
                    String curDate = DateFormatUtil.toDate(window.getStart());
                    for (TradeSkuOrderBean orderBean : input) {
                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setCurDate(curDate);
                        out.collect(orderBean);
                    }
                }
            }
        );
        // reduceDS.print(">>>");
        //TODO 11.关联商品sku维度
        /*
        //11.1最基本的维度关联实现
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                private Connection hbaseConn;

                @Override
                public void open(Configuration parameters) throws Exception {
                    hbaseConn = HbaseUtil.getHbaseConnection();
                }

                @Override
                public void close() throws Exception {
                    HbaseUtil.closeHbaseConnection(hbaseConn);
                }

                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    //根据流中的对象获取要关联的维度的主键
                    String skuId = orderBean.getSkuId();
                    //根据维度的主键获取对应的维度对象
                    //id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                    JSONObject dimJsonObj = HbaseUtil.getDimInfoFromHbase(hbaseConn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", skuId);
                    //将维度对象的相关属性补充到流中的对象上
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    return orderBean;
                }
            }
        );
        //11.2 在基本维度关联的基础上加了旁路缓存优化
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                private Connection hbaseConn;
                private Jedis redisClient;

                @Override
                public void open(Configuration parameters) throws Exception {
                    hbaseConn = HbaseUtil.getHbaseConnection();
                    redisClient = RedisUtil.getRedisClient();
                }

                @Override
                public void close() throws Exception {
                    HbaseUtil.closeHbaseConnection(hbaseConn);
                    RedisUtil.closeRedisClient(redisClient);
                }

                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    //根据流中的对象获取要关联的维度的主键
                    String skuId = orderBean.getSkuId();
                    //根据维度的主键获取对应的维度对象
                    //id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                    JSONObject dimJsonObj = DimUtil.getDimInfo(redisClient,hbaseConn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", skuId);
                    //将维度对象的相关属性补充到流中的对象上
                    if(dimJsonObj != null){
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    }

                    return orderBean;
                }
            }
        );

        //11.3 在旁路缓存的基础上 引入模板方法设计模式
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new DimMapFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            }
        );

        //11.4 发送异步请求关联sku维度---没有使用模板
        // 将异步 I/O 操作应用于 DataStream 作为 DataStream 的一次转换操作
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            //如果发送异步请求 实现分发请求的 AsyncFunction
            new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {

                private StatefulRedisConnection<String, String> asyncRedisConn;
                private AsyncConnection asyncHbaseConn;
                @Override
                public void open(Configuration parameters) throws Exception {
                    asyncRedisConn = RedisUtil.getAsyncRedisConnection();
                    asyncHbaseConn = HbaseUtil.getAsyncConnection();
                }

                @Override
                public void close() throws Exception {
                    RedisUtil.closeAsyncRedisConnection(asyncRedisConn);
                    HbaseUtil.closeAsyncConnection(asyncHbaseConn);
                }

                //发送异步请求，进行维度关联
                @Override
                public void asyncInvoke(TradeSkuOrderBean orderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {

                    //先根据主键到Redis缓存中获取对应的维度数据---异步
                    JSONObject dimJsonObj = RedisUtil.asyncGetDimInfo(asyncRedisConn,"dim_sku_info:"+orderBean.getSkuId());
                    //如果从redis中没有找到对应的维度，发送请求到hbase中查询维度数据--异步
                    if(dimJsonObj == null){
                        dimJsonObj = HbaseUtil.getDimInfoFromHbaseByAsync(asyncHbaseConn, GmallConfig.HBASE_NAMESPACE,"dim_sku_info",orderBean.getSkuId());
                        //并将查询出的维度放到Redis中缓存起来 --异步
                        RedisUtil.asyncWriteDim(asyncRedisConn,"dim_sku_info:"+orderBean.getSkuId(),dimJsonObj);
                    }
                    if(dimJsonObj != null){
                        //将维度相关的属性补充到流中的对象---异步
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    }
                    //将补充后的维度发送到下游去 获取数据库交互的结果并发送给 ResultFuture 的 回调 函数
                    resultFuture.complete(Collections.singleton(orderBean));
                }
            },
            60,
            TimeUnit.SECONDS
        );
        */
        //11.5 发送异步请求关联sku维度  ----使用了模板
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            },
            60,
            TimeUnit.SECONDS
        );
        // withSkuInfoDS.print(">>>");

        //TODO 12.关联商品spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
            withSkuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setSpuName(dimJsonObj.getString("spu_name"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSpuId();
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 13.关联商品tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
            withSpuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setTrademarkName(dimJsonObj.getString("tm_name"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getTrademarkId();
                }
            },
            60,
            TimeUnit.SECONDS
        );
        // withTmDS.print(">>>");
        //TODO 14.关联商品category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream.unorderedWait(
            withTmDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory3Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory3Name(dim.getString("name"));
                    bean.setCategory2Id(dim.getString("category2_id"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 15.关联商品category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream.unorderedWait(
            c3Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory2Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory2Name(dim.getString("name"));
                    bean.setCategory1Id(dim.getString("category1_id"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 16.关联商品category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> restulStream = AsyncDataStream.unorderedWait(
            c2Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory1Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory1Name(dim.getString("name"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        restulStream.print(">>>");

        //TODO 17.将结果写到Doris中
        restulStream
            .map(new BeanToJsonStrFunction<>())
            .sinkTo(DorisUtil.getDorisSink("dws_trade_sku_order_window"));

        env.execute();
    }
}
